package com.sugon.ohdfs.integration.flink.job.stream.sink;

import com.sugon.ohdfs.integration.flink.domain.TestItem;
import com.sugon.ohdfs.integration.flink.service.TestItemSource;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.io.Serializable;
import java.nio.charset.StandardCharsets;

public class RowEncodedStreamSinkJob extends AbstractStreamSinkJob{

    @Value("${hdfs.base-path:hdfs://10.11.8.29:9001/flink/sinkTest/stream/sink/row}")
    String basePath;

    @Autowired
    TestItemSource testItemSource;

    @Override
    protected Serializable transform(TestItem item) {
        return item.toString();
    }

    @Override
    protected StreamingFileSink buildSink() {
        return StreamingFileSink.forRowFormat(new Path(basePath), new SimpleStringEncoder<String>(StandardCharsets.UTF_8.name()))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                .withRolloverInterval(1000*10)
                                .withInactivityInterval(1000*10)
                                .withMaxPartSize(1024*1024)
                                .build()
                )
                .build();
    }


}
